#ifndef __MOYING_COMMU_REQUEST_H__
#define __MOYING_COMMU_REQUEST_H__
#include <moying_comm_lib/moying_commu_base.h>
#include <moying_comm_lib/receive_callback.h>
#include <memory>
#include <mutex>
#include <queue>
#include <moying_comm_lib/waiter.h>
namespace moying
{
namespace communication
{

struct ReqData // 请求数据结构
{
  std::string content_;
  bytes_ptr data_;
  RequestCallbackBasePtr func_;
  bool log_;

  ReqData(const std::string &content, bytes_ptr data, RequestCallbackBasePtr func, bool log) : content_(content), data_(data), func_(func), log_(log)
  {
  }
};

class MoyingCommuRequest : public MoyingCommuBase
{
public:
  MoyingCommuRequest(const std::string &url, const std::string &identity);
  MoyingCommuRequest(const std::string &ip, const unsigned int port, const std::string &identity);
  ~MoyingCommuRequest() noexcept;
  
  //全局函数做为回调
  template <typename REQ, typename RES>
  bool requestCB(const std::string &content, REQ req, void(*req_cb)(const REQ&, const RES&), bool discardBeforeConnected = false, bool log = true)
  {
    return request<REQ, RES>(content, req, std::make_shared<RequestCallback<REQ, RES>>(req_cb), discardBeforeConnected, log);
  }
  
  //lamda或std::function对象做为回调 
  template <typename REQ, typename RES>
  bool request(const std::string &content, REQ req, std::function<void(const REQ&, const RES&)> req_cb, bool discardBeforeConnected = false, bool log = true)
  {
    return request<REQ, RES>(content, req, std::make_shared<RequestCallback<REQ, RES>>(req_cb), discardBeforeConnected, log);
  }

  //类成员函数做为回调
  template<typename REQ, typename RES, class T>
  bool request(const std::string& content, REQ req, void(T::*fp)(const REQ&, const RES&), T* obj, bool discardBeforeConnected = false, bool log = true)
  {
    return request<REQ, RES>(content, req, std::make_shared<RequestCallback<REQ, RES>>(boost::bind(fp, obj, _1, _2)), discardBeforeConnected, log);
  }

  //只发送请求不处理应答
  template<typename REQ>
  bool sendRequest(const std::string& content, const REQ &req, bool discardBeforeConnected = true, bool log = true)
  {
    bytes_ptr pack_data = packProtobufData(content, req);
    if(discardBeforeConnected && is_connected_ == false){
        callLog(MoyingCommuLogLevels::INFO,"%s","discard send request before connected");
        return false; //连接成功前丢充掉所有请求
    }

    bool is_empty = false;
    {
      std::lock_guard<std::mutex> lk(mutex_send_queue_);
      is_empty = queue_send_.empty();
      queue_send_.emplace(content, pack_data, nullptr, log);
    }
    if (is_empty)
      wait_send_queue_.broadcast();
    return true;
  }

  //发送请求后,阻塞等待应答
  template<typename REQ, typename RES>
  bool requestAndWait(const std::string& content, const REQ &req, RES &res, unsigned int wait_mill = 0, bool log = true)
  {
    if(is_waiting_.find(content) != is_waiting_.end())
      return false;
    bool ret = request<REQ, RES>(content, req, 
              std::make_shared<RequestCallback<REQ, RES>>(boost::bind(&MoyingCommuRequest::internalRequestWaitCallback<REQ, RES>, this, content, _1, _2)),
              true, log);
    if(!ret)
      return false;
    std::chrono::system_clock::time_point start_time = std::chrono::system_clock::now();
    is_waiting_.emplace(content);
    ret = false;
    while(wait_mill == 0 || std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now() - start_time).count() < wait_mill)
    {
      if (waited_responses_.find(content) != waited_responses_.end())
      {
        std::lock_guard<std::mutex> lk(mutex_send_queue_);
        ::google::protobuf::Message *res_ptr = &res;
        if(!res_ptr->ParseFromString(waited_responses_[content]))
        {
            printf("%s\n","sub_recv_func parse protobuf failed in getProtobufMessage()");
        }
        waited_responses_.erase(content);
        ret = true;
        break;
      }
      SLEEP(1);
    }
    is_waiting_.erase(content);
    return ret;
  }

  virtual bool run() override;
  virtual bool close() override;
  virtual void setStatusCall(StatusCall func) override;
  virtual void setDiscardPacketBeforeConntect(bool param) override;

private:
  template<typename REQ, typename RES>
  bool internalRequestWaitCallback(std::string content, const REQ &req, const RES &res)
  {
    const ::google::protobuf::Message *msg = &res;
    std::string msg_str;
    msg->SerializeToString(&msg_str);
    {
      std::lock_guard<std::mutex> lk(mutex_send_queue_);
      waited_responses_[content] = msg_str;
    }
    return true;
  }

  template<class P>
  bytes_ptr packProtobufData(const std::string& content, P raw_data)
  {
    unsigned long int timestamp = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count();
    return packProtobufData(content, timestamp, raw_data);
  }

  template<class P>
  bytes_ptr packProtobufData(const std::string& content, unsigned long int timestamp, P raw_data)
  {
    // ::google::protobuf::Message *msg = &raw_data;
    std::string msg_str;
    // msg->SerializeToString(&msg_str);
    raw_data.SerializeToString(&msg_str);

    moying_proto::ServicePack pack;
    pack.set_topic(content);
    // pack.set_timestamp(timestamp);
    pack.set_request(msg_str);
    
    std::string pack_str;
    pack.SerializeToString(&pack_str);
    return std::make_shared<bytes_vec>(pack_str.data(), pack_str.data() + pack_str.length());
  }

  template<typename REQ, typename RES>
  bool request(const std::string &content, REQ req, RequestCallbackBasePtr cb, bool discardBeforeConnected, bool log)
  {
    if(proto_types_.find(content) == proto_types_.end())
    {
      proto_types_[content] = std::make_pair<::google::protobuf::Message*, ::google::protobuf::Message*>(new REQ(), new RES());
    }
    bytes_ptr pack_data = packProtobufData(content, req);
    if(discardBeforeConnected && is_connected_ == false){
        callLog(MoyingCommuLogLevels::INFO,"%s","discard send request before connected");
        return false; //连接成功前丢充掉所有请求
    }

    bool is_empty = false;
    {
      std::lock_guard<std::mutex> lk(mutex_send_queue_);
      is_empty = queue_send_.empty();
      queue_send_.emplace(content, pack_data, cb, log);
    }
    if (is_empty)
      wait_send_queue_.broadcast();
    return true;
  }

  void createLoop();
  void clearSendQueue();
  void printPackWrapper(bool is_send, const std::string &content, bytes_ptr pack, int thread_id);
  void invokeRecvCall(bytes_ptr data);
  bool restart();

private:
  void status_cb(MoyingCommuStatus status, const std::string &msg);

private:
//  zmq::context_t ctx_;   // 上下文环境
  zmq::socket_t socket_; // 套接字对象
  bool  is_connected_   = false;    //是否连接上服务端
  bool  is_timeout_     = false;    //是否超时
  bool  discard_packet_before_connect_  = false;    //连接建立前是否丢弃请求包
  bool  disconnect_after_timeout_    = false;    //是否是请求超时后重新连接引起的连接断开
//  bool  is_restarting_;  //正在执行restart

  std::mutex mutex_send_queue_;    // 发送队列互斥量
  std::queue<ReqData> queue_send_; // 待发送数据队列
  Waiter wait_send_queue_;         // 等待发送队列非空
  std::set<std::string> is_waiting_;
  std::map<std::string, std::string> waited_responses_;
  std::map<std::string, std::pair<RequestCallbackBasePtr, bool>> response_callbacks_;
  std::mutex mutex_waited_responses_;

  StatusCall user_define_status_cb_ = nullptr;  //用户定义的状态回调函数

  std::map<std::string, std::pair<::google::protobuf::Message*, ::google::protobuf::Message*>> proto_types_;
};

typedef std::shared_ptr<MoyingCommuRequest> MoyingCommuRequestPtr;

} // namespace communication
} // namespace moying

#endif // __MOYING_COMMU_REQUEST_H__
